package com.hdu.dwh.config;

import com.esotericsoftware.minlog.Log;
import com.hdu.dwh.enums.HttpStatusEnum;
import com.hdu.dwh.exception.CustomException;
import com.hdu.dwh.properties.FlinkProperties;
import com.hdu.dwh.properties.MinioProperties;
import lombok.Data;
import org.apache.flink.configuration.DeploymentOptions;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.CatalogDescriptor;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.context.annotation.RequestScope;

@Data
@Configuration
public class FlinkConfig {
    @Bean
    @RequestScope
    public TableEnvironment tableEnvironment(FlinkProperties flinkProperties, MinioProperties minioProperties){
        // 配置远程Flink集群的JobManager地址和端口
        org.apache.flink.configuration.Configuration configuration = new org.apache.flink.configuration.Configuration();
        configuration.set(JobManagerOptions.ADDRESS, flinkProperties.getAddress());
        configuration.set(JobManagerOptions.PORT, flinkProperties.getPort());
        configuration.set(DeploymentOptions.TARGET, flinkProperties.getTarget());
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inBatchMode()
                .withConfiguration(configuration)
                .build();
        // StreamExecutionEnvironment
        // TableEnvironment tableEnvironment = TableEnvironment.create(settings);
        TableEnvironment tableEnvironment = StreamTableEnvironment.create(
                StreamExecutionEnvironment.createRemoteEnvironment(flinkProperties.getAddress(), flinkProperties.getPort()).setParallelism(2),
                settings
        );
        /**
         * 初始化 flink 的 paimon 连接 catalog
         */
        try {
            // 参数设置
            configuration.setString("type", "paimon");
            configuration.setString("warehouse", flinkProperties.getWarehouse());
            configuration.setString("metastore", "filesystem");
            configuration.setString("s3.endpoint", minioProperties.getEndpoint());
            configuration.setString("s3.access-key", minioProperties.getAccessKey());
            configuration.setString("s3.secret-key", minioProperties.getSecretKey());
            configuration.setString("s3.path.style.access", "true");
            // 创建Catalog描述符
            CatalogDescriptor catalogDescriptor = CatalogDescriptor.of(flinkProperties.getCatalog(), configuration);
            // 创建Catalog
            String sql = "CREATE CATALOG " + flinkProperties.getCatalog() + " WITH ( 'type' = 'paimon', " +
                    "'warehouse' = '" + flinkProperties.getWarehouse() + "', " +
                    "'metastore' = 'filesystem', " +
                    "'s3.endpoint' = '" + minioProperties.getEndpoint() + "', " +
                    "'s3.access-key' = '" + minioProperties.getAccessKey() + "', " +
                    "'s3.secret-key' = '" + minioProperties.getSecretKey() + "', " +
                    "'s3.path.style.access' = 'true' )";
            tableEnvironment.executeSql(sql);
            tableEnvironment.useCatalog(flinkProperties.getCatalog());
        } catch (Exception e) {
            Log.info(e.getMessage());
            e.printStackTrace();
            throw new CustomException(HttpStatusEnum.FLINK_ERROR);
        }
        return tableEnvironment;
    }

    public StreamExecutionEnvironment streamExecutionEnvironment(FlinkProperties flinkProperties){
        org.apache.flink.configuration.Configuration configuration = new org.apache.flink.configuration.Configuration();
        configuration.set(JobManagerOptions.ADDRESS, flinkProperties.getAddress());
        configuration.set(JobManagerOptions.PORT, flinkProperties.getPort());
        configuration.set(DeploymentOptions.TARGET, flinkProperties.getTarget());
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inBatchMode()
                .withConfiguration(configuration)
                .build();
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment(flinkProperties.getAddress(), flinkProperties.getPort());

        return env;
    }
}
